package com.poetic.order;

import com.poetic.order.domain.OrderEvent;
import com.poetic.order.domain.ReceiptEvent;
import com.poetic.order.function.TxPayMatchByJoin;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * <pre>
 *  双流join实时对账
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class TxMatchByJoin {
    public static void main(String[] args) throws Exception {


        String path = "C:\\indigo\\poetic\\flink\\user-behavior-anylysis\\data\\OrderLog.csv";

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        // 读取订单事件流
        KeyedStream<OrderEvent, String> orderEventStream = env.readTextFile(path)
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String data) throws Exception {
                        String[] line = data.split("\\,");

                        return new OrderEvent(Long.valueOf(line[0]),
                                line[1],
                                line[2],
                                Long.valueOf(line[3]));
                    }
                })
                .filter(new FilterFunction<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return StringUtils.isNotBlank(value.getTxId());
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                            @Override
                            public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                return element.getEventTime() * 1000;
                            }
                        })).keyBy(new KeySelector<OrderEvent, String>() {
                    @Override
                    public String getKey(OrderEvent value) throws Exception {
                        return value.getTxId();
                    }
                });
        String receiptPath = "C:\\indigo\\poetic\\flink\\user-behavior-anylysis\\data\\ReceiptLog.csv";
        DataStreamSource<String> receiptEventStream = env.readTextFile(receiptPath);
//        DataStreamSource<String> receiptEventStream = env.socketTextStream("localhost", 9999);
        KeyedStream<ReceiptEvent, String> receiptEventDataStream = receiptEventStream.map(new MapFunction<String, ReceiptEvent>() {
            @Override
            public ReceiptEvent map(String data) throws Exception {

                String[] dataArr = data.split("\\,");
                return new ReceiptEvent(dataArr[0].trim(), dataArr[1].trim(), Long.valueOf(dataArr[2].trim()));
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<ReceiptEvent>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                .withTimestampAssigner(new SerializableTimestampAssigner<ReceiptEvent>() {
                    @Override
                    public long extractTimestamp(ReceiptEvent element, long recordTimestamp) {
                        return element.getEventTime() * 1000;
                    }
                })).keyBy(new KeySelector<ReceiptEvent, String>() {
            @Override
            public String getKey(ReceiptEvent value) throws Exception {
                return value.getTxId();
            }
        });

        SingleOutputStreamOperator<Tuple2<OrderEvent, ReceiptEvent>> processedStream = orderEventStream.intervalJoin(receiptEventDataStream)
                .between(Time.seconds(-5), Time.seconds(5))
                .process(new TxPayMatchByJoin());
        processedStream.print();
        env.execute("tx pay match by join job");

    }
}
